package com.guaoran.source.zookeeper.distributed.lock;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.utils.ZookeeperFactory;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.omg.PortableInterceptor.SYSTEM_EXCEPTION;

import java.io.IOException;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * @author : guaoran
 * @Description : <br/>
 * 分布式锁：
 *  多个节点以有序节点形式共同注册到根节点下，每个节点都去监听比自己小的节点，
 * @date :2018/10/30 8:29
 */
public class DistributedLock implements Watcher, Lock {

    public static void main(String[] args) throws IOException {
        CountDownLatch countDownLatch = new CountDownLatch(5);
        for (int i = 0; i < 5; i++) {
            new Thread(()->{
                try {
                    countDownLatch.await();
                    DistributedLock distributedLock = new DistributedLock();
                    //获得锁
                    distributedLock.lock();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            },"Thread-"+i).start();
            countDownLatch.countDown();
        }
        System.in.read();
    }


    private ZooKeeper zk = null;
    //定义根节点
    private String ROOT_LOCK = "/locks";
    //定义等待前一个锁
    private String WAIT_LOCK;
    //表示当前的锁
    private String CURRENT_LOCK;
    private CountDownLatch countDownLatch;

    public DistributedLock() {
        try {
            zk = new ZooKeeper("192.168.45.131:2181", 4000, this);
            //校验根节点是否存在，如果不存在则创建一个持久化节点
            Stat stat = zk.exists(ROOT_LOCK,false);
            if(stat == null){
                zk.create(ROOT_LOCK,"0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }

        }catch (Exception e){
            e.printStackTrace();
        }
    }
    /**
     * 处理watcher的监听
     * @param event
     */
    @Override
    public void process(WatchedEvent event) {
        if(this.countDownLatch != null){
            this.countDownLatch.countDown();
        }
    }
    @Override
    public boolean tryLock() {
        try {
            //为当前调用创建临时有序节点
            CURRENT_LOCK =
                    zk.create(ROOT_LOCK+"/","0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println(Thread.currentThread().getName()+"->"+CURRENT_LOCK+"->尝试竞争锁");
            List<String> childrens = zk.getChildren(ROOT_LOCK,false);
            //定义一个集合进行排序
            SortedSet<String> sortedSet = new TreeSet<>();
            for (String children: childrens) {
                sortedSet.add(ROOT_LOCK+"/"+children);
            }
            //获得当前排序集合中的子节点最小的节点
            String firstNode = sortedSet.first();
            SortedSet<String> lessThenMe = sortedSet.headSet(CURRENT_LOCK);
            //如果当前节点与集合中的第一个元素是相同的，即，是最小的节点，获得锁成功
            if (CURRENT_LOCK.equals(firstNode)) {
                return true;
            }
            //获得比当前节点更小的最后一个节点，设置给WAIT_LOCK
            if(!lessThenMe.isEmpty()){
                //将自己前面的  节点设置为自己要等待的锁，
                //只有该节点删除后才能获得锁
                WAIT_LOCK = lessThenMe.last();
            }


        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }
    @Override
    public void lock() {
        //获得锁成功
        if(tryLock()){
            System.out.println(Thread.currentThread().getName()+"->"+CURRENT_LOCK+"->获得锁成功");
            return;
        }
        //如果未获得锁，则监听等待获得锁
        try {
            waitForLock(WAIT_LOCK);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private  boolean waitForLock(String prev) throws KeeperException, InterruptedException {
        //监听当前节点的上一个节点
        Stat stat = zk.exists(prev,true);
        if(stat != null){
            System.out.println(Thread.currentThread().getName()+"->等待锁"+prev+"释放");
            countDownLatch = new CountDownLatch(1);
            countDownLatch.await();
            System.out.println(Thread.currentThread().getName()+"->获得锁");
        }
        return true;
    }
    @Override
    public void lockInterruptibly() throws InterruptedException {

    }



    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        System.out.println(Thread.currentThread().getName()+"->释放锁"+CURRENT_LOCK);
        try {
            zk.delete(CURRENT_LOCK,-1);
            CURRENT_LOCK = null;
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public Condition newCondition() {
        return null;
    }


}
